package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"github.com/influxdata/influxdb/client/v2"
	"io"
	"log"
	"net/http"
	url2 "net/url"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"
)

type Reader interface {
	Read(rc chan []byte)
}
type Writer interface {
	Write(wc chan *Message)
}

type ReadFromFile struct {
	path string
}

type WriteToInfluxDB struct {
	influxDBDsn string
}

type LogProcess struct {
	rc chan []byte
	wc chan *Message

	read  Reader
	write Writer
}

type Message struct {
	TimeLocal                    time.Time
	BytesSent                    int
	Path, Method, Scheme, Status string
	UpstreamTime, RequestTime    float64
}

const (
	TypeHandleLine = 0
	TypeErrNum     = 1
)

var TypeMonitorChan = make(chan int, 200)

// 系统状态监控
type SystemInfo struct {
	HandleLine   int     `json:"handleLine"`   // 总处理日志行数
	Tps          float64 `json:"tps"`          // 系统吞吐量
	ReadChanLen  int     `json:"readChanLen"`  // read channel 长度
	WriteChanLen int     `json:"writeChanLen"` // write channel 长度
	RunTime      string  `json:"runTime"`      // 运行总时间
	ErrNum       int     `json:"errNum"`       // 错误数
}
type Monitor struct {
	startTime time.Time
	data      SystemInfo
	tpsSli    []int
}

func (m *Monitor) start(lp *LogProcess) {
	go func() {
		for n := range TypeMonitorChan {
			switch n {
			case TypeErrNum:
				m.data.ErrNum += 1
			case TypeHandleLine:
				m.data.HandleLine += 1
			}
		}
	}()

	ticker := time.NewTicker(time.Second * 5)
	go func() {
		for {
			<-ticker.C
			m.tpsSli = append(m.tpsSli, m.data.HandleLine)
			if len(m.tpsSli) > 2 {
				m.tpsSli = m.tpsSli[1:]
			}
		}
	}()
	http.HandleFunc("/monitor", func(writer http.ResponseWriter, request *http.Request) {
		m.data.RunTime = time.Now().Sub(m.startTime).String()
		m.data.ReadChanLen = len(lp.rc)
		m.data.WriteChanLen = len(lp.wc)

		if len(m.tpsSli) >= 2 {
			m.data.Tps = float64(m.tpsSli[1]-m.tpsSli[0]) / 5
		}

		ret, _ := json.MarshalIndent(m.data, "", "\t")

		io.WriteString(writer, string(ret))
	})
	http.ListenAndServe(":9193", nil)

}

// read
func (r *ReadFromFile) Read(rc chan []byte) {

	f, err := os.Open(r.path)
	if err != nil {
		panic(fmt.Sprintf("open file error:%s", err.Error()))
	}

	// 从文件尾部开始逐行读取文件内容
	f.Seek(0, 2)
	rd := bufio.NewReader(f)

	for {
		line, err := rd.ReadBytes('\n')
		if err == io.EOF {
			time.Sleep(500 * time.Millisecond)
			continue
		} else if err != nil {
			panic(fmt.Sprintf("ReadBytes error:%s", err.Error()))
		}
		TypeMonitorChan <- TypeHandleLine

		rc <- line[:len(line)-1]
	}

}

//write
func (w *WriteToInfluxDB) Write(wc chan *Message) {
	//for v:= range wc{
	//	fmt.Println(v)
	//}
	/////////////////////////////////////////////////////////////////
	infSli := strings.Split(w.influxDBDsn, "@")
	// Create a new HTTPClient
	c, err := client.NewHTTPClient(client.HTTPConfig{
		Addr:     infSli[0],
		Username: infSli[1],
		Password: infSli[2],
	})
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()
	// Create a new point batch
	bp, err := client.NewBatchPoints(client.BatchPointsConfig{
		Database:  infSli[3],
		Precision: infSli[4],
	})
	if err != nil {
		log.Fatal(err)
	}
	for v := range wc {
		// Create a point and add to batch
		// Tags: Path, Method, Scheme, Status
		tags := map[string]string{"Path": v.Path, "Method": v.Method, "Scheme": v.Scheme, "Status": v.Status}
		// Fields: UpstreamTime, RequestTime, BytesSent
		fields := map[string]interface{}{
			"UpstreamTime": v.UpstreamTime,
			"RequestTime":  v.RequestTime,
			"BytesSent":    v.BytesSent,
		}
		// fmt.Println(fields)
		pt, err := client.NewPoint("nginx_log", tags, fields, v.TimeLocal)
		if err != nil {
			log.Fatal(err)
		}
		bp.AddPoint(pt)
		// Write the batch
		if err := c.Write(bp); err != nil {
			log.Fatal(err)
		}
		// Close client resources
		if err := c.Close(); err != nil {
			log.Fatal(err)
		}
		log.Println("write success!")
	}
}

// process
func (l *LogProcess) Process() {
	// "-" |  "0.008" | "0.008" | 127.0.0.1 | - | [19/OCT/2018:23:30:57 +0800] | "GET /AUTH/LOGIN HTTP/1.1" | 200 | 3552 | "HTTP://NEW.MOYOKEJI.COM/"  | "MOZILLA/5.0 (MACINTOSH; INTEL MAC OS X 10_14_0) APPLEWEBKIT/537.36 (KHTML, LIKE GECKO) CHROME/69.0.3497.100 SAFARI/537.36"
	//'"$http_x_forwarded_for" |  "$request_time" | "$upstream_response_time" | $remote_addr | $remote_user | [$time_local] | "$request" | '
	//'$status | $body_bytes_sent | "$http_referer"  | "$http_user_agent" ';

	r := regexp.MustCompile(`\"([^"]+)\"\s+\|\s+\"([^"]+)\"\s+\|\s+([\d\.]+)\s+\|\s([^ \[]+)\s+\|\s+\[(.*)\]\s+\|\s+\"(.*?)\"\s+\|\s+([\d]+)\s+\|\s+([\d]+)\s+\|\s+\"(.*?)\"\s+\|\s+\"(.*?)\"`)

	loc, _ := time.LoadLocation("Asia/Shanghai")
	//loc, _ = time.LoadLocation("Local")
	log.Println(loc)
	// 循环读取内容
	for v := range l.rc {

		ret := r.FindStringSubmatch(string(v))
		//log.Println(len(ret))
		//log.Println(ret)
		//log.Println("=========================================================")
		//log.Println(ret[0])
		//log.Println(ret[4])
		//log.Println("=========================================================")
		if len(ret) != 11 {
			TypeMonitorChan <- TypeHandleLine
			log.Println("FindStringSubmatch fail:", string(v))
			continue
		}
		message := &Message{}
		//log.Println(ret[5])
		t, err := time.ParseInLocation("02/Jan/2006:15:04:05 -0700", ret[5], loc)
		if err != nil {
			log.Println("ParseInLocation fail:", err.Error(), ret[5])
			continue
		}
		message.TimeLocal = t

		byteSent, _ := strconv.Atoi(ret[8])
		message.BytesSent = byteSent

		// GET /foo?query=t HTTP/1.0
		reqSli := strings.Split(ret[6], " ")

		if len(reqSli) != 3 {
			TypeMonitorChan <- TypeHandleLine
			log.Println("strings Split fail:", ret[6])
			continue
		}
		message.Method = reqSli[0]
		url, err := url2.Parse(reqSli[1])
		if err != nil {
			TypeMonitorChan <- TypeHandleLine
			log.Println("url parse fail:", err.Error())
			continue
		}
		message.Path = url.Path

		// 协议：http
		//message.Scheme = ret[5]
		message.Scheme = "http"
		message.Status = ret[7]

		upstreamTime, _ := strconv.ParseFloat(ret[2], 64)
		requestTime, _ := strconv.ParseFloat(ret[1], 64)
		message.UpstreamTime = upstreamTime
		message.RequestTime = requestTime
		l.wc <- message

		//for _, v := range ret{
		//	fmt.Println(v)
		//}
		//l.wc <- ""
		//ret := r.FindStringSubmatch(string(v))
		//fmt.Println(ret)
		//l.wc<- strings.ToUpper(string(v))
	}

}

/*
create database logProcess
select * from nginx_log
*/
func main() {

	var path, influxDsn string
	flag.StringVar(&path, "path", "/var/log/nginx/access.log", "read file path")
	flag.StringVar(&influxDsn, "influxDsn", "http://127.0.0.1:8086@@@logProcess@s", "influx data source")
	flag.Parse()

	//fmt.Println(path)
	//fmt.Println(influxDsn)
	//return

	r := &ReadFromFile{
		// path:"/var/log/nginx/access.log",
		path: path,
	}
	w := &WriteToInfluxDB{
		//influxDBDsn: "http://127.0.0.1:8086@imooc@imoocpwd@logProcess@s",
		influxDBDsn: influxDsn,
	}
	lp := &LogProcess{
		rc:    make(chan []byte),
		wc:    make(chan *Message),
		read:  r,
		write: w,
	}
	go lp.read.Read(lp.rc)
	go lp.Process()
	go lp.write.Write(lp.wc)

	//
	m := &Monitor{
		startTime: time.Now(),
		data:      SystemInfo{},
	}
	m.start(lp)

}
